package com.ultrapower.hbase;

import java.util.Enumeration;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;

import com.ultrapower.config.Config;
import com.ultrapower.config.ConfigMapping;

/**
 * 
 * @author wangy
 *
 */
public class HbaseConfigurationFactory {
	private final static Set<ConfigMapping<?>> HBASE_CONFIG = new HashSet<ConfigMapping<?>>();
	private Properties properties;
	Configuration configuration = null;

	public HbaseConfigurationFactory() {
		properties = Config.getPropertiesByMapping(HBASE_CONFIG);
	}

	static {
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.client.host", "hbase.zookeeper.quorum", ""));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.client.port", "hbase.zookeeper.property.clientPort", ""));
		HBASE_CONFIG.add(new ConfigMapping<String>("base.zookeeper.znode.parent", "zookeeper.znode.parent", "/hbase"));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.ipc.client.tcpnodelay", "", ""));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.rpc.timeout", "", ""));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.client.operation.timeout", "", ""));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.ipc.client.socket.timeout.read", "", ""));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.ipc.client.socket.timeout.write", "", ""));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.client.async.enable", "", "false"));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.client.async.in.queuesize", "", "10000"));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.tablemultiplexer.flush.period.ms", "", "100"));
		HBASE_CONFIG.add(new ConfigMapping<String>("hbase.client.max.retries.in.queue", "", "10"));
	}

	public Configuration getHbaseConfig() {
		configuration = configuration==null ? HBaseConfiguration.create() : configuration;
		addProperties(configuration, properties);
		return configuration;
	}

	private void addProperties(Configuration configuration, Properties properties) {
		if (properties != null) {
			Enumeration<?> props = properties.propertyNames();
			while (props.hasMoreElements()) {
				String key = props.nextElement().toString();
				configuration.set(key, properties.getProperty(key));
			}
		}
	}
}
